package com.atguigu.transform;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.ArrayList;

/**
 * Created by RUI on 2021/6/9 16:18
 */
public class TransformTest {

    StreamExecutionEnvironment env;

    @Before
    public void before(){
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",2000);
        env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
    }

    @After
    public void after(){
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void mapTest(){

        DataStreamSource<Integer> dss1 = env.fromElements(1, 2, 3, 6, 4, 8, 7);

        dss1.map(new MapFunction<Integer, Integer>() {
            @Override
            public Integer map(Integer value) throws Exception {
                return value * value;
            }
        }).print();


        dss1.map(elments->elments * elments).print();

    }

    @Test
    public void richMapTest(){

        DataStreamSource<Integer> dss1 = env.fromElements(1, 2, 3, 6, 4, 8, 7);

        dss1
            .map(new RichMapFunction<Integer, Integer>() {

                @Override
                public void open(Configuration parameters) throws Exception {
                    System.out.println("open...");
                }

                @Override
                public Integer map(Integer value) throws Exception {
                    return value * value;
                }

                @Override
                public void close() throws Exception {
                    System.out.println("close...");
                }
            })
            .print();


        dss1.map(elments->elments * elments).print();

    }

    @Test
    public void flatMapTest(){
        DataStreamSource<Integer> dss1 = env.fromElements(1, 2, 3, 6, 4, 8, 7);
        dss1.flatMap(new FlatMapFunction<Integer, Integer>() {
            @Override
            public void flatMap(Integer value, Collector<Integer> out) throws Exception {
                if (value % 2 ==0)
                    out.collect(value);
            }
        }).print();
    }

    @Test
    public void reduceTest(){
        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

        DataStreamSource<WaterSensor> dss = env.fromCollection(waterSensors);
        KeyedStream<WaterSensor, String> keyedStream = dss.keyBy(WaterSensor::getId);
/*        keyedStream.reduce(new ReduceFunction<WaterSensor>() {
            @Override
            public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                return new WaterSensor(value1.getId(),null,value1.getVc()+value2.getVc());
            }
        }).print();*/

        keyedStream.reduce((v1,v2)->new WaterSensor(v1.getId(),v1.getTs(),v1.getVc()+v2.getVc())).print();

    }

}
